package com.chao.mail.api.message

import com.chao.mail.api.dto.MailSendMessage
import com.chao.mail.api.message.topic.SEND
import com.chao.mail.application.MailApplicationService
import com.google.gson.Gson
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean
import org.springframework.kafka.annotation.KafkaListener
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.stereotype.Component

@ConditionalOnBean(value = [KafkaTemplate::class])
@Component
class MailCommandReceiver{
    private val logger = LoggerFactory.getLogger(MailCommandReceiver::class.java)
    private val gson = Gson()
    @Autowired
    lateinit var mailApplicationService: MailApplicationService

    @KafkaListener(topics = [SEND])
    fun mailSendRequestReceive(kafkaRecord: ConsumerRecord<String, String>){
        try {
            val string: String = kafkaRecord.value()
            logger.info("receive new send request: $string")
            val mailSendMessage = gson.fromJson(string, MailSendMessage::class.java)
            mailSendMessage?.let {
                mailApplicationService.sendMail(mailSendMessage)
            }
        }catch (e: Exception){
            logger.error(e.message)
        }
    }
}